package mn.idax.exchange.otc.service;

import com.alibaba.fastjson.JSON;
import mn.idax.exchange.auth.bean.AuthInfo;
import mn.idax.exchange.auth.interfaces.AuthService;
import mn.idax.exchange.auth.util.AuthHelper;
import mn.idax.exchange.common.SpringUtils;
import mn.idax.exchange.otc.coder.JsonSerializer;
import mn.idax.exchange.otc.config.RedisSubListenerConfig;
import mn.idax.exchange.otc.dto.*;
import mn.idax.exchange.otc.entity.ChatLog;
import mn.idax.exchange.otc.entity.OTCArbitration;
import mn.idax.exchange.otc.entity.OTCOrder;
import mn.idax.exchange.otc.entity.UserInfo;
import mn.idax.exchange.otc.enums.LogTypeEnum;
import mn.idax.exchange.otc.enums.MessageTypeEnum;
import mn.idax.exchange.otc.util.AuthUtils;
import mn.idax.exchange.otc.util.FileUtils;
import mn.idax.exchange.otc.util.MessageNotifyUtils;
import org.apache.commons.lang.StringUtils;
import org.atmosphere.config.service.*;
import org.atmosphere.cpr.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springsource.loaded.C;

import javax.inject.Inject;
import java.io.IOException;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 客户、客服主动聊天
 */

@ManagedService(path = "/order/{chatRoom:[a-zA-Z][a-zA-Z_0-9]*}")
public class ChatRoomService {

    private static final Logger logger = LoggerFactory.getLogger(ChatRoomService.class);
    private static final String CHAT_ROOM_ATTRIBUTES = "chatRoomAttributes";
//    private final static String AUTH_USER = "user";
//    private final static String CAN_CHAT = "canChat";
//    private final static String BUYERID = "buyerId";
//    private final static String SELLID = "sellId";
//    private static String USER_LANGUAGE = "language";

    @Inject
    public static BroadcasterFactory broadcasterFactory;

    private final ConcurrentHashMap<String, String> users = new ConcurrentHashMap<>();
    public static final ConcurrentHashMap<String, ConnectionDTO> CONNECTIONS = new ConcurrentHashMap<>();

    /**
     * chat room id is order id
     **/
    @PathParam("chatRoom")
    private String chatRoom;

    /**
     * client chat connect onOpen
     *
     * @param resource
     */
    @Get
    public void onOpen(final AtmosphereResource resource) {
        resource.addEventListener(new AtmosphereResourceEventListenerAdapter() {
            @Override
            public void onSuspend(AtmosphereResourceEvent event) {
                resource.setSerializer(new JsonSerializer());
            }
        });
    }

    /**
     * client chat connect onReady
     *
     * @param resource
     */
    @Ready()
    public void onReady(final AtmosphereResource resource) {

        logger.info("----onReady start,if you seeing this####----");
        logger.info("----connect resource uuid:{}, orderId/chatRoom:{}----", resource.uuid(), chatRoom);

        /**check order state**/
        ChatLogService chatLogService = SpringUtils.getBean(ChatLogService.class);
        OTCOrder order = chatLogService.getOTCOrderByOrderId(chatRoom);
        if (order == null) {
            logger.warn("----order not found:{}----", chatRoom);
            this.closeResource(resource);
            return;
        }

        if (order.getOrderstate().intValue() == 4 ||
                order.getOrderstate().intValue() == 6 ||
                order.getOrderstate().intValue() == 9 ||
                order.getOrderstate().intValue() == 10) {
            logger.info("----order state is closed:{}----", order.getOrderstate().intValue());
            this.closeResource(resource);
            return;
        }

        /**save userId to session**/
        ChatRoomAttributes chatRoomAttributes = new ChatRoomAttributes();


        /**find broadcaster for this chat room, create if null**/
        Broadcaster broadcaster = broadcasterFactory.lookup(chatRoom, true);

        //带token的 老的连接
        /**check client token**/
        if (!Objects.isNull(resource.getRequest().getParameter("token"))){
            AuthInfo authInfo = AuthUtils.getAuthInfoByToken(resource.getRequest());
            if (authInfo == null || authInfo.getId() <= 0) {
                this.closeResource(resource);
                return;
            }

            int senderId = authInfo.getId();


            //--app 上后 删除
            //标识用户是否可以参与聊天
            boolean canJoin = false;
            boolean canChat = false;
            boolean isCustomerServiceStaff = false;

            //是否交易双方
            canJoin = chatLogService.isRelatedUser(chatRoom, senderId);

            //是否客服人员
            if (!canJoin) {
                canJoin = chatLogService.checkIsStaffByUserId(senderId);
                isCustomerServiceStaff = canJoin;
            }

            if (canJoin) {
                canChat = true;
            }

            //查询是否是后台人员 user source = 0
            if (authInfo.getSource() == 0) {
                canJoin = true;
            }

            if (!canJoin) {
                logger.warn("----not trading user and staff----");
                this.closeResource(resource);
                return;
            }

            chatRoomAttributes.setUser(senderId);
            chatRoomAttributes.setCanChat(canChat);

            /**user connect**/
            if (!users.containsKey(String.valueOf(senderId))) {
                users.put(String.valueOf(senderId), resource.uuid());
            } else {
                logger.info("----connect sender:{}, already connect----", senderId);
            }
            logger.info("----####staff join chat userId-{} isStaff-{} canJoin-{}####----", senderId, isCustomerServiceStaff, canJoin);

            if (isCustomerServiceStaff) {

                logger.info("----####staff join chat send notify ####----");
                chatRoomAttributes.setStaff(senderId);
                chatRoomAttributes.setStaff_order(chatRoom);

                customerJoinedChat(chatRoomAttributes,resource);

            }

            /**add AtmosphereResource to broadcaster**/
            broadcaster.addAtmosphereResource(resource);

        } else {
            //将该连接信息丢到内存里
            ConnectionDTO connectionDTO = new ConnectionDTO(System.currentTimeMillis(),resource);
            CONNECTIONS.put(resource.uuid(),connectionDTO);
            logger.info("onReadyChatRoomAttributes params {}",chatRoomAttributes);
        }

        chatRoomAttributes.setBuyerId(order.getBuyer());
        chatRoomAttributes.setSellId(order.getSeller());
        chatRoomAttributes.setOrderNum(order.getNumber());

        String lang = resource.getRequest().getParameter("lang");
        logger.info("onReadyLang is {}",lang);
        if (!StringUtils.isEmpty(lang)){
            chatRoomAttributes.setLanguage(lang);
        } else {
            chatRoomAttributes.setLanguage("zh_cn");

        }
        resource.getRequest().getSession().setAttribute(CHAT_ROOM_ATTRIBUTES, chatRoomAttributes);

    }

    /**
     * when client send a message, call this method
     *
     * @param resource
     * @param message
     */
    @Message
    public void onMessage(AtmosphereResource resource, String message) {

        if (Objects.isNull(resource.getRequest().getParameter("token"))){
            //建立连接成功后的第一条消息 传送内容为token 不广播出去
            if (!firstMessageSaveUserInfo(resource,message)){
                return;
            }
        }

        logger.info("onMessage start{},{}", resource.uuid(), chatRoom);

        if (StringUtils.isEmpty(message)) {
            return;
        }

        if (!isUserCanChat(resource)) {
            return;
        }

        /**get user info from session**/
        int senderId = getAuthUserId(resource);
        if (senderId == -1) {
            return;
        }

        /**build chatProtocol with request message, and check it**/
        ChatProtocol chatProtocol = JSON.parseObject(message, ChatProtocol.class);

        if (!isValidChatProtocol(chatProtocol)) {
            return;
        }

        /**text message too long**/
        if (isMessageTooLong(chatProtocol)) {
            return;
        }

        String imgAddress = "";
        FileUrlService fileUrlService = SpringUtils.getBean(FileUrlService.class);
        /**when messageType is file, get file info from redis, rebuild chatProtocol**/
        if (chatProtocol.getMsgType().equals(MessageTypeEnum.file)
                || chatProtocol.getMsgType().equals(MessageTypeEnum.image)
                || chatProtocol.getMsgType().equals(MessageTypeEnum.audio)
                || chatProtocol.getMsgType().equals(MessageTypeEnum.video)
                ) {
            ChatRedisService chatRedisService = SpringUtils.getBean(ChatRedisService.class);
            Object object = chatRedisService.getChatFileInfo(chatProtocol.getMessage());
            if (Objects.isNull(object)) {
                return;
            }
            //删除缓存
            chatRedisService.deleteChatFileInfo(chatProtocol.getMessage());

            ChatFileDto chatFileDto = JSON.parseObject(object.toString(), ChatFileDto.class);
            chatProtocol.setName(chatFileDto.getName());

            chatProtocol.setSource(fileUrlService.subFilePath(chatFileDto.getSource()));
            chatProtocol.setMsgType(getMessageTypeEnum(chatFileDto.getName()));

            imgAddress = chatFileDto.getSource();
        }

        /**get broadcaster for broadcast**/
        //Broadcaster broadcaster = getCurrBroadcaster(resource);

        /**1:package chatProtocol before saving to mongo **/
        chatProtocol = packageChatProtocol(senderId, chatProtocol);

        /**2:save chatLog successfully, then broadcast**/
        if (chatProtocol != null) {
            ChatRoomAttributes attribute = (ChatRoomAttributes) resource.getRequest().getSession(false).getAttribute(CHAT_ROOM_ATTRIBUTES);
            chatProtocol.setBuyerReadFlag(attribute.getUser().equals(attribute.getBuyerId()) ? 1 : 0);
            chatProtocol.setSellerReadFlag(attribute.getUser().equals(attribute.getSellId()) ? 1 : 0);
            chatProtocol.setBuyerId(attribute.getBuyerId());
            chatProtocol.setSellerId(attribute.getSellId());
            chatProtocol.setOrderNum(attribute.getOrderNum());

            logger.info("onMessageSaveChatLogToMongo chatProtocol params {}",chatProtocol);
            ChatLog chatLog = saveChatLogToMongo(chatProtocol);
            if (chatLog == null) {
                logger.warn("----save chat log to mongo failure:{}----", JSON.toJSONString(chatProtocol));
                return;
            }
            /**3:send message to related client**/
            chatProtocol.setSaved(true);
            chatProtocol.setSource(imgAddress);

            SpringUtils.getBean(ChatRedisService.class).pushMessageToRedis(RedisSubListenerConfig.OTC_ORDER_STATE_CHANGE_EVENT, JSON.toJSONString(chatProtocol));
        } else {
            logger.warn("----package ChatProtocol failure:{}----", JSON.toJSONString(chatProtocol));
        }
    }

    @Disconnect
    public void onDisconnect(AtmosphereResourceEvent event) {
        logger.info("----onDisconnect start,chatRoom:{}----", chatRoom);
        if (event.isCancelled()) {
            users.values().remove(event.getResource().uuid());
            logger.info("----disconnect resource uuid:{}----", event.getResource().uuid());
        } else if (event.isClosedByClient()) {
            logger.info("----disconnect by client, resource uuid:{}----", event.getResource().uuid());
        }
    }


    /**
     * 建立连接后第一次发送信息 信息内容为user token
     * @param resource
     * @param message
     * @return
     */
    public boolean firstMessageSaveUserInfo(AtmosphereResource resource, String message){

        logger.info("firstMessageSaveUserInfoMessage data {}",message);
        ChatRoomAttributes chatRoomAttributes = (ChatRoomAttributes)resource.getRequest().getSession(false).getAttribute(CHAT_ROOM_ATTRIBUTES);

        //第一次连接
        if (Objects.isNull(chatRoomAttributes.getUser())){
            if (StringUtils.isEmpty(message)){
                return false;
            }

            /**check client token**/
            Map<String,String> map = JSON.parseObject(message, Map.class);
            logger.info("firstMessageSaveUserInfoToken data {}",JSON.toJSONString(map));
            String token = map.get("token");
            AuthHelper authHelper = AuthHelper.getInstance(SpringUtils.getBean(AuthService.class));
            if (!authHelper.isTokenValid(token)) {
                return false;
            }

            AuthInfo authInfo = authHelper.decodeToken(token);
            if (authInfo == null || authInfo.getId() <= 0) {
                this.closeResource(resource);
                return false;

            }
            int senderId = authInfo.getId();

            //标识用户是否可以参与聊天
            boolean canJoin = false;
            boolean canChat = false;
            boolean isCustomerServiceStaff = false;

            ChatLogService chatLogService = SpringUtils.getBean(ChatLogService.class);
            //是否交易双方
            canJoin = chatLogService.isRelatedUser(chatRoom, senderId);

            //是否客服人员
            if (!canJoin) {
                canJoin = chatLogService.checkIsStaffByUserId(senderId);
                isCustomerServiceStaff = canJoin;
            }

            if (canJoin) {
                canChat = true;
            }

            //查询是否是后台人员 user source = 0
            if (authInfo.getSource() == 0) {
                canJoin = true;
            }

            if (!canJoin) {
                logger.warn("----not trading user and staff----");
                this.closeResource(resource);
                return false;
            }

            /**save userId to session**/
            chatRoomAttributes.setUser(senderId);
            chatRoomAttributes.setCanChat(canChat);

            /**find broadcaster for this chat room, create if null**/
            Broadcaster broadcaster = broadcasterFactory.lookup(chatRoom, true);

            /**add AtmosphereResource to broadcaster**/
            broadcaster.addAtmosphereResource(resource);

            /**user connect**/
            if (!users.containsKey(String.valueOf(senderId))) {
                users.put(String.valueOf(senderId), resource.uuid());
            } else {
                logger.info("----connect sender:{}, already connect----", senderId);
            }
            logger.info("----####staff join chat userId-{} isStaff-{} canJoin-{}####----", senderId, isCustomerServiceStaff, canJoin);

            if (isCustomerServiceStaff) {
                chatRoomAttributes.setStaff(senderId);
                chatRoomAttributes.setStaff_order(chatRoom);
                logger.info("----####staff join chat send notify ####----");
                customerJoinedChat(chatRoomAttributes,resource);
            }

            /**save userId to session**/
            chatRoomAttributes.setUser(senderId);
        }
        ConnectionDTO connectionDTO = CONNECTIONS.get(resource.uuid());
        connectionDTO.setTimestamp(0);
        return true;
    }

    /**
     * 客服加入聊天
     * @param chatRoomAttributes
     * @param resource
     */
    public void customerJoinedChat(ChatRoomAttributes chatRoomAttributes,AtmosphereResource resource){

        //仲裁加入新字段标识客服进入过
        OTCArbitrationService otcArbitrationService = SpringUtils.getBean(OTCArbitrationService.class);

        /**
         * 客服第一次加入：持久化消息&发送消息&记录客服已介入
         * 客服后续加入：只推送临时消息
         * 客服每次退出：只推送临时消息
         */

        //如果是客服并且第一次进入此仲裁订单 推送消息+持久化
        boolean customerFirstJoined = true;

        OTCArbitration otcArbitration = otcArbitrationService.queryArbitrationByUniqueKey(chatRoom);
        if (otcArbitration != null && otcArbitration.getCustomerjoined()) {
            customerFirstJoined = false;
        }

        if (customerFirstJoined) {
            MessageNotifyUtils.joinStaffNotify(resource,chatRoomAttributes);
            //更新仲裁 标识客服不是第一次进入
            int update = otcArbitrationService.updateCustomerJoined(otcArbitration.getOrderid());
            logger.info("----updateOtcArbitration tabel finish,Influence numbers {}", update);
        }
    }

    /**
     * packageChatProtocol
     *
     * @param senderId
     * @param chatProtocol
     * @return
     */
    private ChatProtocol packageChatProtocol(Integer senderId, ChatProtocol chatProtocol) {
        try {
            if (chatProtocol != null && senderId != -1) {

                /**set order id**/
                chatProtocol.setOrderId(chatRoom);

                /**get order & user info from session**/
                OrderUserDto orderUserDto = SpringUtils.getBean(ChatLogService.class).getUserByOrderId(chatProtocol.getOrderId());
                if (orderUserDto == null) {
                    logger.warn("----packageChatProtocol order info not found, senderId:{}, chatProtocol{}----", senderId, JSON.toJSONString(chatProtocol));
                    return null;
                }

                UserInfo userInfo = SpringUtils.getBean(ChatLogService.class).getUserInfoByOrderIdAndUserId(chatProtocol.getOrderId(), senderId);
                if (userInfo == null) {
                    userInfo = SpringUtils.getBean(ChatLogService.class).getUserInfoById(senderId);
                    if (userInfo == null) {
                        logger.warn("----packageChatProtocol user info not found, senderId:{}, chatProtocol{}----", senderId, JSON.toJSONString(chatProtocol));
                        return null;
                    } else {
                        logger.warn("----packageChatProtocol user is staff, staff id:{}, chatProtocol{}----", senderId, JSON.toJSONString(chatProtocol));
                    }
                }

                /**set LogType**/
                LogTypeEnum logType = null;
                if (orderUserDto.getBuyer().intValue() == senderId) {
                    logType = LogTypeEnum.buyer;
                } else if (orderUserDto.getSeller().intValue() == senderId) {
                    logType = LogTypeEnum.seller;
                } else {
                    logType = LogTypeEnum.staff;
                }

                /**add extra info**/
                return ChatProtocol.addExtraInfo(chatProtocol, userInfo, logType);
            }
        } catch (Exception e) {
            logger.error("----package chat protocol failure----", JSON.toJSONString(chatProtocol), e);
        }
        return null;
    }

    /**
     * saveChatLogToMongo
     *
     * @param chatProtocol
     * @return
     */
    private ChatLog saveChatLogToMongo(ChatProtocol chatProtocol) {
        try {
            if (chatProtocol != null) {
                ChatLog chatLog = new ChatLog();
                BeanUtils.copyProperties(chatProtocol, chatLog);
                logger.info("::::SaveChatLog::ChatRoom", JSON.toJSONString(chatLog));
                return SpringUtils.getBean(ChatLogService.class).insert(chatLog);
            }
        } catch (Exception e) {
            logger.error("----save chatLog to mongo error----", e);
        }
        return null;
    }

    /**
     * getAuthUserId by resource(from session)
     *
     * @param resource
     * @return
     */
    private int getAuthUserId(AtmosphereResource resource) {
        try {
            return Integer.parseInt(((ChatRoomAttributes)resource.getRequest().getSession(false).getAttribute(CHAT_ROOM_ATTRIBUTES)).getUser().toString());
        } catch (Exception e) {
            return -1;
        }
    }


    /**
     * isUserCanChat by resource(from session)
     *
     * @param resource
     * @return
     */
    private boolean isUserCanChat(AtmosphereResource resource) {
        try {
//            return Boolean.valueOf(((ChatRoomAttributes)resource.getRequest().getSession(false).getAttribute(CHAT_ROOM_ATTRIBUTES)).getCanChat().toString());
            return ((ChatRoomAttributes)resource.getRequest().getSession(false).getAttribute(CHAT_ROOM_ATTRIBUTES)).isCanChat();
        } catch (Exception e) {
            return false;
        }
    }

    /**
     * closeResource
     *
     * @param resource
     */
    private void closeResource(AtmosphereResource resource) {
        try {
            resource.close();
        } catch (IOException e) {
            logger.warn("----close connect error, connect resource uuid:{}----", resource.uuid());
        }
    }

    /**
     * check chatProtocol
     *
     * @param chatProtocol
     * @return
     */
    private boolean isValidChatProtocol(ChatProtocol chatProtocol) {

        if (chatProtocol == null || StringUtils.isEmpty(chatProtocol.getMessage()) || chatProtocol.getMsgType() == null) {
            return false;
        }

        if (MessageTypeEnum.text.equals(chatProtocol.getMsgType()) || MessageTypeEnum.file.equals(chatProtocol.getMsgType())
                || MessageTypeEnum.image.equals(chatProtocol.getMsgType()) || MessageTypeEnum.audio.equals(chatProtocol.getMsgType())
                || MessageTypeEnum.video.equals(chatProtocol.getMsgType())
                ) {
            return true;
        }

        logger.warn("----ChatProtocol is invalid : {}----", JSON.toJSONString(chatProtocol));
        return false;
    }

    /**
     * add UserAndRoom To ChatProtocol
     *
     * @param chatProtocol
     * @return
     */
    private ChatProtocol addUserAndRoomToChatProtocol(ChatProtocol chatProtocol) {
        chatProtocol.setUsers(users.keySet());
        chatProtocol.setRooms(chatRoom);
        return chatProtocol;
    }

    /**
     * getCurrBroadcaster by resource
     *
     * @param resource
     * @return
     */
    private Broadcaster getCurrBroadcaster(AtmosphereResource resource) {
        Broadcaster broadcaster = broadcasterFactory.lookup(chatRoom, true);
        broadcaster.addAtmosphereResource(resource);
        return broadcaster;
    }

    /**
     * @param chatProtocol
     * @return
     */
    private boolean isMessageTooLong(ChatProtocol chatProtocol) {
        if (chatProtocol != null && StringUtils.isNotEmpty(chatProtocol.getMessage())) {
            return chatProtocol.getMessage().getBytes().length > 1000;
        }
        return false;
    }

    /**
     * @param fileName
     * @return
     */
    private MessageTypeEnum getMessageTypeEnum(String fileName) {
        String extName = fileName.substring(fileName.lastIndexOf(".") + 1).toLowerCase();
        FileUtils fileUtils = SpringUtils.getBean(FileUtils.class);
        Object object = fileUtils.getType().get(extName);
        if (object == null) {
            return null;
        }
        return MessageTypeEnum.valueOf(object.toString());
    }

}
